package com.li.sparkstreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 通过socket模拟产生数据，实时计算单词出现次数
 */
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf();
    conf.setMaster("local[2]"); //local[2]表示启动两个进程，1个读取数据，一个处理数据
    conf.setAppName("StreamWordCount");

    //创建StreamContext,指定数据处理间隔5s
    val streamingContext = new StreamingContext(conf, Seconds(5))

    //通过sockct获取实时产生的数据
    val linesRDD = streamingContext.socketTextStream("bigdata04", 9001)

    //使用空格切割转换为单个单词
    val wordRDD = linesRDD.flatMap(_.split(" "))

    //把单词转为tuple2
    val tupRDD = wordRDD.map((_, 1))

    //求和
    val wordCountRDD = tupRDD.reduceByKey(_ + _)

    wordCountRDD.print()

    //启动
    streamingContext.start()
    //等待停止
    streamingContext.awaitTermination()
  }
}
